package com.yu.boot2.rocket;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @author zy
 * @date 2019-01-31 14:35
 **/
@Slf4j
@Service
public class ReceiveService extends DefaultConsumerConfigure implements ApplicationListener<ContextRefreshedEvent> {

    @Override
    public void onApplicationEvent(ContextRefreshedEvent arg0) {
        try {
            super.listener("TopicTest", "Tag1");
        } catch (MQClientException e) {
            log.error("消费者监听器启动失败", e);
        }

    }

    @Override
    public ConsumeConcurrentlyStatus dealBody(List<MessageExt> msgs) {
        int num = 1;
        log.info("进入");
        for (MessageExt msg : msgs) {
            log.info("第" + num + "次消息");
            String msgStr = new String(msg.getBody(), StandardCharsets.UTF_8);
            // TODO 处理消息
            log.info(msgStr);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
